package com.buddy.zookeeper.master;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Random;

public class Master implements Watcher{
    ZooKeeper zk;
    String hostPort;
    String serverId = Integer.toString(new Random().nextInt());
    static boolean isLeader = false;


    public Master(String hostPort) {
        this.hostPort = hostPort;
    }

    void startZK() throws IOException {
        zk = new ZooKeeper(hostPort,15000,this);
    }
    void stopZK() throws InterruptedException {
        zk.close();
    }

    boolean checkMaster() throws KeeperException, InterruptedException {
        while (true){
            try {
                Stat stat = new Stat();
                byte[] data = zk.getData("/master",false,stat);
                isLeader = new String(data).equals(serverId);
                return true;
            } catch (KeeperException.NoNodeException e) {
                e.printStackTrace();
                return false;
            } catch (KeeperException.ConnectionLossException e) {
                e.printStackTrace();
            }
        }
    }
    void runForMaster() throws InterruptedException, KeeperException {
        while (true) {
            try {
                zk.create("/master", serverId.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                isLeader = true;
                break;
            } catch (KeeperException.NoNodeException e) {
                e.printStackTrace();
                isLeader = false;
                break;
            }catch (KeeperException.NodeExistsException e){
                e.printStackTrace();
                isLeader = false;
                break;
            }catch (KeeperException.ConnectionLossException e){
                e.printStackTrace();
            }
            if (checkMaster()) break;
        }
    }
    public void process(WatchedEvent watchedEvent) {
        System.out.println(watchedEvent);
    }

    public static void main(String[] args) throws InterruptedException, IOException, KeeperException {
        Master master = new Master("127.0.0.1:2181");
        master.startZK();
        master.runForMaster();
        if (isLeader){
            System.out.println("I'm the leader");
            Thread.sleep(60000);
        }else {
            System.out.println("Someone else is the leader");
        }
        master.stopZK();
    }
}
